package com.sheepone.distributor;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sheepone.distributor.strategy.BusyWaitStrategy;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Miss.杨
 * @description
 * @since 2024/1/26 - 14:16
 */
public abstract class DisruptorHolder<T> {

    protected long lastRecord = 0;

    protected EventFactory<T> factory;

    protected String name;

    protected AtomicBoolean resizing = new AtomicBoolean(false);

    protected Disruptor<T> line;

    protected EventHandler<T> handler;

//    protected BusyWaitStrategy waitStrategy;

    public DisruptorHolder(String name, EventFactory<T> factory, EventHandler<T> handler) {
        this.name = name;
        this.factory = factory;
        this.handler = handler;
    }

    public DisruptorHolder(String name, EventFactory<T> factory) {
        this.name = name;
        this.factory = factory;
    }

    public abstract void stop();

    public long hasPublishedNum() {
        return this.lastRecord + line.getCursor() + 1;
    }

//    public void onRetry() {
//        waitStrategy.onRetry();
//    }
//
//    public void onResize() {
//        waitStrategy.onResize();
//    }


    /**
     * 判断是否正在扩容
     *
     * @return
     */
    public boolean isResizing() {
        return this.resizing.get();
    }

    public void resize(int size) {
        if (this.resizing.compareAndSet(false, true)) {
            this.line.halt();
            this.lastRecord = hasPublishedNum();
            this.line = newDisruptor(size);
            this.line.start();
            resizing.set(false);
        }
    }

    public String getName() {
        return name;
    }

    protected Disruptor<T> newDisruptor(int size) {
        Disruptor<T> disruptor = new Disruptor<>(
                factory,// 事件工厂
                size, // 大小
                r -> { // 线程工厂
                    Thread thread = new Thread(r);
                    thread.setName(name);
                    thread.setDaemon(true);
                    return thread;
                },
                ProducerType.MULTI,// 生产类型
                new YieldingWaitStrategy()// 等待策略
        );
        disruptor.handleEventsWith(handler);
        return disruptor;
    }
}
